package com.vbobot.demo.rabbitmq;

import com.rabbitmq.client.Channel;
import java.io.IOException;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.context.annotation.Scope;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

/**
 * @author Bobo
 * @date 2022/11/9
 */
@Slf4j
@Component
@Scope("singleton")
public class MessageListener {
    public static AtomicInteger SLEEP_SECONDS = new AtomicInteger(100);

    @RabbitListener(
            bindings = @QueueBinding(
                    value = @Queue(RabbitConstats.QUEUE),
                    exchange = @Exchange(value = RabbitConstats.EXCHANGE, type = ExchangeTypes.TOPIC),
                    key = RabbitConstats.ROUTING_KEY
            ),
            concurrency = "4"
    )
    public void onMessage(@Payload Map<String, Object> body, Message message) {
        final Object messageId = body.get("value");
        log.info("receive:{}, redelivered:{}", messageId, message.getMessageProperties().getRedelivered());
        try {
            Thread.sleep(SLEEP_SECONDS.get());
        } catch (InterruptedException e) {
            log.info("throw error, message:{}, error:{}", messageId, e.getMessage());
        }
        log.info("finish:{}", messageId);
    }

    @RabbitListener(
            bindings = @QueueBinding(
                    value = @Queue(RabbitConstats.MANUAL_QUEUE),
                    exchange = @Exchange(value = RabbitConstats.EXCHANGE_MANUAL, type = ExchangeTypes.TOPIC),
                    key = RabbitConstats.ROUTING_KEY
            ),
            concurrency = "4",
            containerFactory = "manualAckRabbitListenerContainerFactory"
    )
    public void onManualMessage(@Payload Map<String, Object> body, Message message, Channel channel) {
        final Object messageId = body.get("value");
        log.info("receive:{}, redelivered:{}", messageId, message.getMessageProperties().getRedelivered());
        boolean ack = true;
        try {
            Thread.sleep(SLEEP_SECONDS.get());
        } catch (InterruptedException e) {
            ack = false;
            log.info("throw error, message:{}, error:{}", messageId, e.getMessage());
        } finally {
            log.info("ack, message:{}, ack:{}", messageId, ack);
            ack(message, channel, ack);
        }
        log.info("finish:{}", messageId);
    }

    @SneakyThrows(IOException.class)
    private static void ack(Message message, Channel channel, Boolean ack) {
        final long deliveryTag = message.getMessageProperties().getDeliveryTag();
        if (ack) {
            channel.basicAck(deliveryTag, false);
        } else {
            channel.basicNack(deliveryTag, false, false);
        }
    }
}
